package com.lm.flink.component;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisSinkFunction extends RichSinkFunction<String> {
    private transient JedisPool jedisPool;
    private final ParameterTool params;

    public RedisSinkFunction(ParameterTool params) {
        this.params = params;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        this.jedisPool = createJedisPool(this.params);
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.select(0);
            jedis.incrBy("flink:counter:" + value, 1);
        }
    }

    @Override
    public void close() throws Exception {
        if (jedisPool != null && !jedisPool.isClosed()) {
            jedisPool.close();
        }
    }

    private JedisPool createJedisPool(ParameterTool params) {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(100);
        poolConfig.setMaxIdle(50);
        poolConfig.setMinIdle(10);
        poolConfig.setTestOnBorrow(true);

        String host = params.get("redis.host", "localhost");
        int port = params.getInt("redis.port", 6379);
        String password = params.get("redis.password");
        int timeout = params.getInt("redis.timeout", 2000);

        if (password != null && !password.isEmpty()) {
            return new JedisPool(poolConfig, host, port, timeout, password);
        } else {
            return new JedisPool(poolConfig, host, port, timeout);
        }
    }
}